package cn.edu.flink.scala.tutorial.state.keyedstate

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object ListStateTest2 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)


    val stream1 = env.fromElements(
      ("a", "stream-1", 1000L),
      ("b", "stream-1", 2000L),
      ("b", "stream-1", 3000L)
    ).assignAscendingTimestamps(_._3)

    val stream2 = env.fromElements(
      ("a", "stream-2", 3000L),
      ("b", "stream-2", 4000L),
    ).assignAscendingTimestamps(_._3)

    stream1.keyBy(_._1)
      .connect(stream2.keyBy(_._1))
      .process(new CoProcessFunction[(String, String, Long), (String, String, Long), String] {
        lazy val stream1State: ListState[(String, String, Long)] = getRuntimeContext.getListState(new ListStateDescriptor[(String, String, Long)]("stream-1", classOf[(String, String, Long)]))
        lazy val stream2State: ListState[(String, String, Long)] = getRuntimeContext.getListState(new ListStateDescriptor[(String, String, Long)]("stream-2", classOf[(String, String, Long)]))


        override def processElement1(value: (String, String, Long), ctx: CoProcessFunction[(String, String, Long), (String, String, Long), String]#Context, out: Collector[String]): Unit = {
          stream1State.add(value)
          println(value)

          import scala.collection.JavaConversions._

          for (right <- stream2State.get()) {
            out.collect(s"${value} => ${right}")
          }
        }

        override def processElement2(value: (String, String, Long), ctx: CoProcessFunction[(String, String, Long), (String, String, Long), String]#Context, out: Collector[String]): Unit = {
          stream2State.add(value)
          println(value)

          import scala.collection.JavaConversions._

          for (left <- stream1State.get()) {
            out.collect(s"${left} <= ${value}")
          }
        }
      })
      .print()

    env.execute("ListStateTest2")
  }
}
